package com.infinitus.autocatlog.domain;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.Properties;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class KafkaConfig {
	
	@Value("${kafka.bootstrap.servers}")
	protected String servers;
	
	@Value("${kafka.acks}")
	protected String acks;
	
	@Value("${kafka.retries}")
	protected String retries;
	
	@Value("${kafka.batch.size}")
	protected String size;
	
	@Value("${kafka.linger.ms}")
	protected String ms;
	
	@Value("${kafka.buffer.memory}")
	protected String memory;
	
	@Value("${kafka.key.serializer}")
	protected String keySerializer;
	
	@Value("${kafka.value.serializer}")
	protected String valueSerializer;

	@Value("${kafka.send.topic}")
	protected String autoTopic;


	public String getAutoTopic() {
		return autoTopic;
	}


	public Properties getProperties(){
		Properties props = new Properties();
		props.put("bootstrap.servers", servers);
		props.put("acks", acks);
		props.put("retries", retries);
		props.put("batch.size", size);
		props.put("linger.ms", ms);
		props.put("buffer.memory", memory);
		props.put("key.serializer", keySerializer);
		props.put("value.serializer", valueSerializer);
		return props;
	}
	
	public static void main(String[] args) throws IOException, ParseException {
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
		
		BufferedReader bufferedReader = new BufferedReader(
				new InputStreamReader(new FileInputStream("D:\\data_count_log.txt"), "utf-8"));
		String str = null;
		String recDate = "2016-11-14";
		Integer eff = 0;
		Integer inEff = 0;
		while ((str=bufferedReader.readLine())!=null){
			String[] arr = str.split("\\|");
			String receiveDateStr = arr[3];
			if(!receiveDateStr.equals(recDate)){
				float f1 = eff;
				float f2 = inEff;
				
				System.out.println(recDate+"\t\t"+eff+"\t\t"+inEff+"\t\t"+f2/f1);
				recDate = receiveDateStr;
				eff = 0;
				inEff = 0;
			}
			int count = Integer.parseInt(arr[1]);
			Date eventDate = sdf.parse(arr[2]);
			Date receiveDate = sdf.parse(receiveDateStr);
			Calendar calendar = Calendar.getInstance();
			calendar.setTime(eventDate);
			calendar.add(Calendar.DATE, 8);
			if(calendar.getTime().after(receiveDate)){
				eff += count;
			}else{
				inEff += count;
			}
		}
		bufferedReader.close();
	}
}
